#pragma once
#include "../common/net.hpp"
#include <unordered_set>
#include <unordered_map>
namespace BRpc
{
    namespace Server
    {
        class RpcTopic
        {
        public:
            using ptr = std::shared_ptr<RpcTopic>;

        public:
            RpcTopic() {}
            void onRequest(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                TopicOptype topic_optype = msg->optype();
                bool ret = true;
                switch (topic_optype)
                {
                // 主题的创建
                case TopicOptype::TOPIC_CREATE:
                    topicCreate(conn, msg);
                    break;
                // 主题的删除
                case TopicOptype::TOPIC_REMOVE:
                    topicDel(conn, msg);
                    break;
                // 主题的订阅
                case TopicOptype::TOPIC_SUBSCRIBE:
                    ret = topicSubscribe(conn, msg);
                    break;
                // 主题的取消订阅
                case TopicOptype::TOPIC_CANCEL:
                    topicCancel(conn, msg);
                    break;
                // 主题消息的发布
                case TopicOptype::TOPIC_PUBLISH:
                    ret = topicPublish(conn, msg);
                    break;
                default:
                    return errorResponse(conn, msg, RCode::RCODE_INVALID_OPTYPE);
                }
                if (!ret)
                    return errorResponse(conn, msg, RCode::RCODE_NOT_FOUND_TOPIC);
                return topicResponse(conn, msg);
            }
            void onShutdown(const BaseConnection::ptr &conn)
            {
                std::vector<Topic::ptr> topics;
                Subscriber::ptr subscriber;
                // 如果连接是一个消息订阅者的话 我们需要在主题中删除这个消息订阅者的关系
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto subscri_it = _subscribers.find(conn);
                    if (subscri_it == _subscribers.end())
                    {
                        return; // 是一个消息发布者 不做任何处理
                    }
                    subscriber = subscri_it->second;
                    for (auto topic : subscriber->_topics)
                    {
                        topics.push_back(_topics[topic]);
                    }
                }
                // 在各个主题中删除订阅者
                for (auto &topic : topics)
                {
                    topic->removeSubscriber(subscriber);
                }
            }

        private:
            void errorResponse(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg, RCode rcode)
            {
                auto msg_rsp = MessageFactory::create<TopicResponse>();
                msg_rsp->setId(msg->id());
                msg_rsp->setMType(MType::RSP_TOPIC);
                msg_rsp->setRCode(rcode);
                conn->send(msg_rsp);
            }
            void topicResponse(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                auto msg_rsp = MessageFactory::create<TopicResponse>();
                msg_rsp->setId(msg->id());
                msg_rsp->setMType(MType::RSP_TOPIC);
                msg_rsp->setRCode(RCode::RCODE_OK);
                conn->send(msg_rsp);
            }

        private:
            void topicCreate(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                // 检查是否已经创建了 已经创建了则什么都不做
                {
                    std::unique_lock<std::mutex> lock(_mutex);

                    auto topic = _topics.find(msg->topicName());
                    if (topic != _topics.end())
                    {
                        DLOG("%s主题已经创建了", msg->topicName().c_str());
                        return;
                    }
                    // 将主题添加进成员里
                    DLOG("创建主题%s", msg->topicName().c_str());
                    auto it = std::make_shared<Topic>(msg->topicName());
                    _topics.insert(std::make_pair(msg->topicName(), it));
                }
            }
            void topicDel(const BaseConnection::ptr &conn, const TopicRequest::ptr msg)
            {
                // 先将把主题信息在订阅了当前主题的订阅者中删除
                auto name = msg->topicName();
                std::unordered_set<Subscriber::ptr> subscribers;

                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto topic = _topics[name];
                    subscribers = topic->_subscribers;
                }

                // 不知道这样会不会有空指针解引用什么的问题 后面看看
                for (auto &x : subscribers)
                {
                    x->delTopic(name);
                }
            }
            bool topicSubscribe(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                // 1. 先找出主题对象，以及订阅者对象
                //    如果没有找到主题--就要报错；  但是如果没有找到订阅者对象，那就要构造一个订阅者
                Topic::ptr topic;
                Subscriber::ptr subscriber;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto topic_it = _topics.find(msg->topicName());
                    if (topic_it == _topics.end())
                    {
                        return false;
                    }
                    topic = topic_it->second;
                    auto sub_it = _subscribers.find(conn);
                    if (sub_it != _subscribers.end())
                    {
                        subscriber = sub_it->second;
                    }
                    else
                    {
                        subscriber = std::make_shared<Subscriber>(conn);
                        _subscribers.insert(std::make_pair(conn, subscriber));
                    }
                }
                // 2. 在主题对象中，新增一个订阅者对象关联的连接；  在订阅者对象中新增一个订阅的主题
                topic->appendSubscriber(subscriber);
                subscriber->appendTopic(msg->topicName());
                return true;
            }
            void topicCancel(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                // 先找到订阅者 和主题
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto topic = _topics.find(msg->topicName());
                    // 主题不存在什么都不做
                    if (topic == _topics.end())
                    {
                        return;
                    }
                    auto subscriber = _subscribers.find(conn);
                    if (subscriber == _subscribers.end())
                    {
                        return;
                    }
                    subscriber->second->delTopic(msg->topicName());
                    topic->second->removeSubscriber(subscriber->second);
                }
            }
            bool topicPublish(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                auto topic = std::shared_ptr<Topic>();
                // 先找到主题
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto topic_it = _topics.find(msg->topicName());
                    if (topic_it == _topics.end())
                    {
                        return false;
                    }
                    topic = topic_it->second;
                }
                // 这样写的话就可以分离锁 让我们的多线程的效率更高
                topic->pushMessage(msg);
                return true;
            }

        private:
            struct Subscriber
            {
                using ptr = std::shared_ptr<Subscriber>;
                std::mutex _mutex;
                BaseConnection::ptr _conn;               // 订阅者的连接
                std::unordered_set<std::string> _topics; // 订阅了的主题名称
                Subscriber(const BaseConnection::ptr &conn) : _conn(conn) {}
                void appendTopic(const std::string &tName)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    {
                        auto topic = _topics.find(tName);
                        if (topic == _topics.end())
                            _topics.insert(tName);
                    }
                }
                void delTopic(const std::string &tName)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    {
                        auto topic = _topics.find(tName);
                        if (topic != _topics.end())
                            _topics.erase(tName);
                    }
                }
            };
            struct Topic
            {
                using ptr = std::shared_ptr<Topic>;
                std::string _topic_name;
                std::mutex _mutex;
                std::unordered_set<Subscriber::ptr> _subscribers;
                Topic(const std::string name) : _topic_name(name) {}
                void appendSubscriber(const Subscriber::ptr &subscriber)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    {
                        auto man = _subscribers.find(subscriber);
                        if (man == _subscribers.end())
                            _subscribers.insert(subscriber); // 不存在才插入
                    }
                }
                void removeSubscriber(const Subscriber::ptr &subscriber)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    {
                        auto man = _subscribers.find(subscriber);
                        if (man != _subscribers.end())
                            _subscribers.erase(subscriber); // 存在才删除
                    }
                }
                void pushMessage(const BaseMessage::ptr &msg)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    {
                        for (auto &subscriber : _subscribers)
                        {
                            subscriber->_conn->send(msg);
                        }
                    }
                }
            };

        private:
            std::unordered_map<std::string, Topic::ptr> _topics;                   // 所有创建了的主题
            std::unordered_map<BaseConnection::ptr, Subscriber::ptr> _subscribers; // 所有订阅者的信息
            std::mutex _mutex;
        };

    }
}